package com.jsy.hadoop.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author: SongyangJi
 * @descrption:
 * @since: 2021/12/22
 */
public class NaturalJoin {

    public static class JoinMapper extends Mapper<LongWritable, Text, Text, Text> {
        private static final String STUDENT_FILENAME = "student.csv";
        private static final String STUDENT_COURSE_FILENAME = "student_course.csv";
        //标签
        private static final String STUDENT_FLAG = "student";
        private static final String STUDENT_COURSE_FLAG = "student_course";

        private final Text outKey = new Text();
        private final Text outValue = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            // 从context中获取文件输入路径
            FileSplit fileSplit = (FileSplit) context.getInputSplit();
            String filePath = fileSplit.getPath().toString();
            String line = value.toString(); //
            //将value分为SID和其他属性
            String[] fields = line.split(",", 2);
            //判断记录来自哪个文件
            if (filePath.contains(STUDENT_FILENAME)) {
                // 行记录形如：200900130604,李新建,女,24.0,32434.0,数学学院,2011

                //fields[0]为SID，fields[1]为其他属性
                outKey.set(fields[0]);
                // 这里只输出姓名
                String name = fields[1].split(",")[0];
                outValue.set(STUDENT_FLAG + " " + name); // 打上标志位以供后面区分
            } else if (filePath.contains(STUDENT_COURSE_FILENAME)) {
                // 行记录形如：200900130985,300023,94.0,100023

                //fields[0]为SID，fields[1]为其他属性
                outKey.set(fields[0]);
                // 这里只需要输出课程编号和分数
                String[] split = fields[1].split(",");
                String cidAndScore = split[0] + "," + split[1];
                outValue.set(STUDENT_COURSE_FLAG + " " + cidAndScore); // 打上标志位以供后面区分
            }
            context.write(outKey, outValue);
        }
    }

    public static class JoinReducer extends Reducer<Text, Text, Text, Text> {
        private static final String STUDENT_FLAG = "student";
        private static final String STUDENT_COURSE_FLAG = "student_course";
        private final Text Value = new Text();
        private String studentName = null;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            List<String> record = new ArrayList<>();
            // 在此例中是一对多的关系,所以如下处理
            for (Text val : values) {
                String[] fields = StringUtils.split(val.toString(), ' ');
                String fileFlag = fields[0];
                //判断记录来自哪个文件，并根据文件格式解析记录。
                if (fileFlag.equals(STUDENT_FLAG)) {
                    studentName = fields[1];
                } else if (fileFlag.equals(STUDENT_COURSE_FLAG)) {
                    record.add(fields[1]);
                }
            }
            //求笛卡尔积
            for (String cidAndScore : record) {
                Value.set(studentName + "," + cidAndScore);
                context.write(key, Value); // 形如 <key = 200900130001, value = 段玉东,300037,77.0>
            }
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Join");
        job.setJarByClass(NaturalJoin.class);

        job.setMapperClass(JoinMapper.class);
        job.setReducerClass(JoinReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileInputFormat.addInputPath(job, new Path(args[1]));
        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

}
